package com.sfx.mq.demo;

import com.sfx.mq.constants.MQConstant;
import com.sfx.mq.constants.TopicProperties;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;


import java.sql.*;
import java.util.List;
import java.util.UUID;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-10-28
 * Time: 12:10
 */
public class RepeatMessageTest {

    @Test
    public void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("repeat-producer-group");
        producer.setNamesrvAddr(MQConstant.NAME_SRV);
        producer.start();
        String keys = UUID.randomUUID().toString();
        Message message = new Message(TopicProperties.REPEAT_TOPIC, null, keys, "消息-1".getBytes());
        SendResult send = producer.send(message);
        System.out.println("发送结果 : " + send.getSendStatus());
        Message repeatMessage = new Message(TopicProperties.REPEAT_TOPIC, null, keys, "消息-repeat".getBytes());
        SendResult repeatSend = producer.send(repeatMessage);
        System.out.println("发送结果 : " + repeatSend.getSendStatus());

        producer.shutdown();
    }

    /**
     * 解决消息重复问题 :
     * 我们设计一个去重表,对消息的唯一key添加唯一索引 or  redis 的setnx
     * 我们每次消费消息的时候,先插入数据库,如果成功则执行业务逻辑 (如果业务逻辑执行报错,则删除这个去重表记录)
     * 如果插入失败,说明消息来过了,直接签收了
     *
     * @throws Exception
     */

    @Test
    public void Consumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("repeat-consumer-group");
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);
        consumer.subscribe(TopicProperties.REPEAT_TOPIC, "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try {
                    MessageExt messageExt = list.get(0);
                    String keys = messageExt.getKeys();
                    Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/cloud_user?serverTimezone=GMT%2B8&useSSL=false", "root", "123456");
                    PreparedStatement statement = connection.prepareStatement("insert into ordertable(orderNo) values('" + keys + "')");
                    int i = statement.executeUpdate();
                    System.out.println("受影响的行数... row=" + i);

                    // 这里执行业务,如果业务报错,还要把记录删除
                    try {
                        System.out.println("消费了消息 message is " + new String(messageExt.getBody()));
                        handleyewu();
                    } catch (Exception e) {
                        // 删除这一条记录
                        e.printStackTrace();
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (SQLException e) {
                    if (e instanceof SQLIntegrityConstraintViolationException) {
                        System.out.println("消息被重复消费了...");
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    throw new RuntimeException(e);
                }
            }
            private void handleyewu() {

            }
        });
        consumer.start();
        // 挂起JVM
        System.in.read();
    }
}